package com.fwmagic.flink.state;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
import org.apache.flink.util.Collector;

import java.util.Properties;

/**
 * OperatorStateAndKeyedState记录
 *
 * 前提：topic有4个分区，Flink并行度为4，保证每个subTask一个分区
 * 效果：checkpoint的目录中会生成8个目录记录状态，
 *      OperateState:有4个文件做了记录,
 *      KeyState:也有4个文件记录，记录的是Flink消费kafka的偏移量
 */
public class OperatorStateAndKeyedStateDemo {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        env.enableCheckpointing(5000);
        //程序异常退出或者人工cancel掉，checkpoint不删除，保留！
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        System.setProperty("HADOOP_USER_NAME", "hadoop");
        env.setStateBackend(new FsStateBackend("hdfs://192.168.62.131:9000/flink/flink-checkpoints/ch2"));

        Properties prop = new Properties();
        prop.setProperty("bootstrap.servers", "192.168.62.131:9092,192.168.62.132:9092,192.168.62.133:9092");
        prop.setProperty("group.id", "gp3");
        //从最早的数据开始消费
        prop.setProperty("auto.offset.reset", "earliest");
        //kafka的消费者不自动提交偏移量
        prop.setProperty("enable.auto.commit", "false");

        String topic = "state1";

        FlinkKafkaConsumer011<String> kafkaSource = new FlinkKafkaConsumer011<>(topic,
                new SimpleStringSchema(),
                prop);
        DataStreamSource<String> dataStream = env.addSource(kafkaSource);

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = dataStream.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String line, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] words = line.split("\\s");
                for (String word : words) {
                    out.collect(Tuple2.of(word, 1));
                }
            }
        });

        SingleOutputStreamOperator<Tuple2<String, Integer>> sumed = wordAndOne.keyBy(0).sum(1);

        sumed.print();

        env.execute("OperatorStateAndKeyedStateDemo");



    }
}
